package org.paraj.prodcons;

import java.util.concurrent.atomic.AtomicLong;

import org.paraj.misc.StopWatch;

public class StatisticsTracingEngineImpl<T> extends SimpleEngine<T> implements StatisticsTracingEngine<T>{
    private AtomicLong processedCount;
    private AtomicLong consumptionMilliseconds;
    protected AtomicLong productionMilliseconds;

    @Override
    public SimpleEngine<T> start(){
        processedCount = new AtomicLong(0);
        productionMilliseconds = new AtomicLong(0);
        consumptionMilliseconds = new AtomicLong(0);
        productionMilliseconds = new AtomicLong(0);

        return super.start();
    }

    @Override
    protected ConsumerThread createConsumerThread(Consumer c){
        return new StatisticsTracingConsumerThread(c);
    }

    @Override
    protected ProducerThread createProducerThread(Producer<T> p){
        return new StatisticsTracingProduceThread(p);
    }

    public Long processedCount(){
        return processedCount.longValue();
    }

    public Double consumptionSeconds(){
        return millisToSeconds(consumptionMilliseconds.longValue());
    }

    public Double productionSeconds(){
        return millisToSeconds(productionMilliseconds.longValue());
    }

    private double millisToSeconds(Long productionMilliseconds){
        return productionMilliseconds.doubleValue() / 1000;
    }

    protected class StatisticsTracingConsumerThread extends ConsumerThread{
        private StopWatch watch;

        public StatisticsTracingConsumerThread(Consumer c){
            super(c);
        }

        public void run(){
            watch = new StopWatch();
            super.run();
        }

        protected void closeConsumer(){
            watch.start();
            try{
                super.closeConsumer();
            } finally {
                consumptionMilliseconds.addAndGet(watch.stop());
            }
        }

        protected void consume(T product){
            watch.start();
            try{
                super.consume(product);
            } finally {
                consumptionMilliseconds.addAndGet(watch.stop());
                processedCount.incrementAndGet();
            }
        }
    }

    protected class StatisticsTracingProduceThread extends ProducerThread{
        private StopWatch watch;

        public StatisticsTracingProduceThread(Producer<T> p){
            super(p);
        }

        public void run(){
            watch = new StopWatch();
            super.run();
        }

        protected T produce() throws Exception{
            watch.start();
            try{
                return super.produce();
            } finally {
                productionMilliseconds.addAndGet(watch.stop());
            }
        }
    }
}
